package com.atguigu.api

import java.util.{Properties, Random}

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011

/**
 * @ClassName SourceTest
 * @Description
 *             1. 从自定义集合中读取数据
 *             2. 从文件中读取数据
 *             3. 从kafka中读取数据
 *             4. 自定义source
 * @Author Mr Yang
 * @Date 2020/8/20 22:17
 * @Version 1.0
 */

case class SensorReading(id: String, timestamp: Long, temperature: Double)

object SourceTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //1.从自定义集合中读取数据
    val stream1 = env.fromCollection(List(
      SensorReading("sensor_1", 1597933289L, 35.5555555),
      SensorReading("sensor_6", 1597933333L, 35.5555555),
      SensorReading("sensor_7", 1597933269L, 35.5555555),
      SensorReading("sensor_10", 1597933209L, 35.5555555)
    ))
    //2. 从文件中读取数据
    val stream2 = env.readTextFile("F:\\work\\FlinkTutorial\\src\\main\\resources\\sensor.txt");

    //env.fromElements(1, 2.0, "hello flink").print()

    //3. 从kafka中读取数据
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")

    val stream3 = env.addSource( new FlinkKafkaConsumer011[String]( "helloKafka", new SimpleStringSchema(), properties ))

    //4. 自定义source
    val stream4 = env.addSource( new SensorSource() )

    stream4.print("stream4").setParallelism(1)
    env.execute()
  }

  case class SensorSource() extends SourceFunction[SensorReading] {

    //定义一个flag，表示数据源正常执行
    var runing:Boolean = true

    //正常生成数据
    override def run(sourceContext: SourceFunction.SourceContext[SensorReading]): Unit = {
      //初始化一个随机数发生器
      val rand = new Random()

      //初始化一组传感器温度数据
      var curTemp = 1.to(10).map(
        i => ("sensor_" + i, 60 + rand.nextGaussian() * 20)
      )

      //用无限循环，产生数据流
      while (runing) {
        //在前一次温度的基础上更新温度值
        curTemp = curTemp.map(
          t => (t._1,t._2 + rand.nextGaussian())
        )
        //获取当前时间戳
        val curTime = System.currentTimeMillis()
        //包装数据并使用sourceContext输出
        curTemp.foreach(
          t => sourceContext.collect( SensorReading(t._1, curTime, t._2) )
        )
        //设置时间间隔
        Thread.sleep(500)
      }
    }

    //取消数据源生成
    override def cancel(): Unit = {
      runing = false
    }
  }
}
